"""
Case Type   : 协议兼容
Case Name   : sha2协议连接成功，创建及修改二级分区表
Create At   : 2024.09.23
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、caching_sha2认证
    4、查询系统未使用端口
    5、设置协议兼容guc参数
    6、连接og侧B库创建分区表及索引
    7、og数据库查询分区信息
    8、合法添加分区
    9、非法添加分区
    10、合法删除分区
    11、非法删除分区
    12、og数据库查询分区信息
    13、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、成功
    8、成功
    9、报错
    10、成功
    11、报错
    12、成功
    13、成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0017'
        self.u_name = 'u_protocol_0017'
        self.tb_name = 'tb_protocol_0017'
        self.ts_name = 'ts_protocol_0017'
        self.inx_name = 'index_protocol_0017'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exec_sql1 = f'''drop tablespace if exists {self.ts_name};
        create tablespace {self.ts_name} relative location '{self.ts_name}';
        drop table if exists {self.tb_name};
        create table {self.tb_name}(c1 int primary key,c2 int,c3 int)\
        with (storage_type=ustore)partition by range(c1) subpartition by \
        range(c2)(partition p1 values less than (100)\
        (subpartition p1_1 values less than (50),\
        subpartition p1_2 values less than (100)),\
        partition p2 values less than (200)(subpartition p2_1 values less \
        than (150),subpartition p2_2 values less than (200)));
        drop index if exists {self.inx_name};
        create index {self.inx_name} on {self.tb_name} (c1) global;
        drop index if exists {self.inx_name}_01;
        create index {self.inx_name}_01 on {self.tb_name} (c2) local;'''
        self.exec_sql2 = f'''alter table {self.tb_name} add partition p3 \
        values less than(300);
        alter table {self.tb_name} add partition p4 values less than(400) \
        (subpartition p4_1 values less than(400));
        alter table {self.tb_name} add partition p5 values less than(500) \
        (subpartition p5_1 values less than(450),subpartition p5_2 values \
        less than(500));
        alter table {self.tb_name} add partition p6 values less than(600) \
        (subpartition p6_1 values less than(510)),add partition p7 values \
        less than(700);
        alter table {self.tb_name} modify partition p6 add subpartition p6_2 \
        values less than(520);
        alter table {self.tb_name} modify partition p6 add subpartition p6_3 \
        values less than(530),modify partition p6 add subpartition p6_4 values \
        less than(540);
        alter table {self.tb_name} add partition p8 values less than(800) \
        tablespace {self.ts_name};
        alter table {self.tb_name} modify partition p6 add subpartition p6_5 \
        values less than(550) tablespace {self.ts_name};'''
        self.exec_sql3 = f'''alter table {self.tb_name} add partition p8 \
        values less than(900);
        alter table {self.tb_name} add partition p9 values less \
        than(100);
        alter table {self.tb_name} add partition p9 values less \
        than(900) (subpartition p9_1 values less than(100),\
        subpartition p9_2 values less than(50));'''
        self.exec_sql4 = f'''alter table {self.tb_name} drop partition p8;
        alter table {self.tb_name} drop partition p1,drop partition p2;'''
        self.exec_sql5 = f'''alter table {self.tb_name} drop partition pnull;
        alter table {self.tb_name} drop subpartition p_null;'''

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} encoding 'utf8' "
                   f"dbcompatibility ='B';")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertEqual(1,
                         sql_res.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:caching_sha2认证;expect:成功---'
        self.log.info(text)
        sql_cmd = (f"select set_caching_sha2_password('{self.u_name}', "
                   f"'{macro.COMMON_PASSWD}', '');")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(1 row)', sql_res, '创建失败')

        text = '----step4:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '-step5:设置协议兼容guc参数;expect:成功-'
        self.log.info(text)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_dolphin_proto = on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'dolphin_server_port ={port}')
        self.log.info(msg)
        self.assertTrue(msg)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-step6:连接og侧B库创建分区表及索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql1, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step7:og数据库查询分区信息;expect:成功--'
        self.log.info(text)
        sql_cmd = (f"use {self.u_name};"
                   f"select relname,boundaries,parttype from pg_partition "
                   f"where parentid in (select parentid from pg_partition "
                   f"where relname = '{self.tb_name}') or parentid in "
                   f"(select oid from pg_partition where parentid in "
                   f"(select parentid from pg_partition where relname "
                   f"= '{self.tb_name}')) order by relname;")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(7 rows)', sql_res)
        lines = [line.strip() for line in sql_res.strip().split('\n')]
        expected_relnames = [
            'p1', 'p1_1', 'p1_2', 'p2', 'p2_1', 'p2_2', 'tb_protocol_0017']
        for i, relname in enumerate(expected_relnames, start=3):
            assert relname in lines[i], f"Expected '{relname}' in line {i}, \
                    but found '{lines[i]}'"

        text = '-step8:合法添加分区;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql2, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step9:非法添加分区;expect:报错--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql3, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('The partition name "p8" is duplicated with '
                      'existing partition name', c_res)

        text = '-step10:合法删除分区;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql4, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step11:非法删除分区;expect:报错--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql5, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('partition "pnull" does not exist', c_res)

        text = '-step12:og数据库查询分区信息;expect:成功--'
        self.log.info(text)
        sql_cmd = (f"use {self.u_name};"
                   f"select relname,boundaries,parttype from pg_partition "
                   f"where parentid in (select parentid from pg_partition "
                   f"where relname = '{self.tb_name}') or parentid in "
                   f"(select oid from pg_partition where parentid in "
                   f"(select parentid from pg_partition where relname "
                   f"= '{self.tb_name}')) order by relname;")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(16 rows)', sql_res)
        lines = [line.strip() for line in sql_res.strip().split('\n')]
        expected_relnames = [
            'p3', 'p3_subpartdefault1', 'p4', 'p4_1', 'p5', 'p5_1', 'p5_2',
            'p6', 'p6_1', 'p6_2', 'p6_3', 'p6_4', 'p6_5', 'p7',
            'p7_subpartdefault1', 'tb_protocol_0017']
        for i, relname in enumerate(expected_relnames, start=3):
            assert relname in lines[i], f"Expected '{relname}' in line {i}, \
            but found '{lines[i]}'"

    def tearDown(self):
        text = '-----step13:清理环境;expect:成功-----'
        self.log.info(text)
        drop_cmd = f"drop database if exists {self.db_name};"
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        drop_cmd1 = f"drop tablespace if exists {self.ts_name};"
        self.log.info(drop_cmd1)
        drop_res1 = self.sh_pri.execut_db_sql(drop_cmd1, dbname='postgres')
        self.log.info(drop_res1)
        drop_cmd2 = f"drop user if exists {self.u_name};"
        self.log.info(drop_cmd2)
        drop_res2 = self.sh_pri.execut_db_sql(drop_cmd2)
        self.log.info(drop_res2)

        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn('DROP ROLE', drop_res2, '清理环境失败')
        self.assertEqual(1,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.assertIn('DROP TABLESPACE', drop_res1, '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
